package com.huitone.smspfm.interfacz.mo.report;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;

import com.huitone.smspfm.interfacz.logger.ExtLogger;
import com.huitone.smspfm.interfacz.mo.SvrNodeClientsManager;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.bin.common.utils.DateTimeUtils;
import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.context.app.AppManager;
import com.huitone.smspfm.context.thread.AppCustomBlockingThreadPool;
import com.huitone.smspfm.context.utils.ContextPropertiesUtils;
import com.huitone.smspfm.interfacz.core.AppStaticConfigs;
import com.huitone.smspfm.interfacz.dao.entity.ErrorMng;
import com.huitone.smspfm.interfacz.dao.entity.SvrNodeDef;
import com.huitone.smspfm.interfacz.model.NotifyHash;
import com.huitone.smspfm.interfacz.model.SmsReport;
import com.huitone.smspfm.redis.model.AppCanstant;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;
import com.huitone.smspfm.socket.client.SmsClient;
import com.huitone.smspfm.socket.protocol.sgip.constant.CommandID;
import com.huitone.smspfm.socket.protocol.sgip.message.Message;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.IBody;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Report;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Report2;
import com.huitone.smspfm.socket.protocol.sgip.message.msghead.Head;
import com.huitone.smspfm.socket.utils.SerialUtils;

/**
 * @Title ReportThreadManager.java
 * @Package com.huitone.smspfm.interfacz.core
 * @Description TODO(用一句话描述该文件做什么)
 * @author ybliang
 * @date 2017年4月4日 上午11:53:23
 * @version V1.0
 */
public class ReportManager extends Thread {
	
	private static final ExtLogger logger = ExtLogger.create(ReportManager.class);
	
	private ExecutorService executorService = null;
	
	private long sleepTime = 1000;
	
	private boolean toSleep = false;
	
	private static final String CurrReportSpeedS = "CurrReportSpeedS";
	//private static final String ReportSpeedTimeS = "ReportSpeedTimeS";
	private static final String CurrReportSpeedM = "CurrReportSpeedM";
	//private static final String ReportSpeedTimeM = "ReportSpeedTimeM";
	
	private static final byte[] CURR_REPORT_SPEEDS_B = AppStaticConfigs.stringRedisSerializer.serialize(CurrReportSpeedS);
	//private static final byte[] REPORT_SPEED_TIMES_B = AppStaticConfigs.stringRedisSerializer.serialize(ReportSpeedTimeS);
	
	private static final byte[] CURR_REPORT_SPEEDM_B = AppStaticConfigs.stringRedisSerializer.serialize(CurrReportSpeedM);
	//private static final byte[] REPORT_SPEED_TIMEM_B = AppStaticConfigs.stringRedisSerializer.serialize(ReportSpeedTimeM);
	
	private byte[] currReportSpeedS_List = AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.RunningParamsHash + ":" + CurrReportSpeedS);
	private byte[] currReportSpeedM_List = AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.RunningParamsHash + ":" + CurrReportSpeedM);
	
	private static long lastTime = System.currentTimeMillis();
	
	private static long lastCount = 0;
	
	private static long mLastTime = System.currentTimeMillis();
	
	private static long mLastCount = 0;
	
	private ArrayBlockingQueue<Map<byte[],byte[]>> speeds = new ArrayBlockingQueue<Map<byte[],byte[]>>(10000);
	
	private static ReportManager instance = new ReportManager();
	
	public static ReportManager getInstance() {
		return instance;
	}

	private ReportManager(){}
	
	@Override
	public void run() {
		
		executorService = AppCustomBlockingThreadPool.newBlockingThreadPool(StringUtils.toInteger(ContextPropertiesUtils.getProperty("smspfm.server.report.poolsize"), 30, true));
		
		AppManager.esRegister.add(executorService);
		
		if (AppStaticConfigs.speedStatisticsOn) {
			speedWriter();
		}
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> script = new DefaultRedisScript<List>(AppStaticConfigs.luaGetRpt, List.class);
		
		List<String> keys = new ArrayList<>();
		keys.add(AppStaticConfigs.iSMNotifyList);
		keys.add(AppStaticConfigs.pSMCacheHash);
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> script2 = new DefaultRedisScript<>(AppStaticConfigs.luaRptSendSucc, List.class);
		
		List<String> keys2 = new ArrayList<>();
		keys2.add(AppStaticConfigs.iSMNotifySerialNoSet);
		keys2.add(AppStaticConfigs.iSMRptRespSerialNoZSet);
		
		while (!this.isInterrupted() && AppManager.isRunning()) {
			
			executorService.execute(new Runnable() {
				
				@Override
				public void run() {
					
					List<List<String>> lists = AppStaticConfigs.scriptOpts3.getHashList(script, keys, AppStaticConfigs.rptReadPerTime + "");
					
					List<SmsReport> entities = new ArrayList<>();
					for (int i = 0 ; i < lists.size() ; i ++) {
						List<String> list = lists.get(i);
						SmsReport entity = SmsReport.parse(list);
						entities.add(entity);
					}
					
					if (entities.isEmpty()) {
						toSleep = true;
						return;
					}
					
					toSleep = false;
					
					List<NotifyHash> nfhs = new ArrayList<NotifyHash>();
					
					AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {
						
						@Override
						public void excute(RedisConnection connection) {
					
							for (SmsReport sr:entities) {
								
								if (sr.isWellInformed()) {
									
									ErrorMng errorMng = ErrorMng.get(sr);
									Integer errorCode = errorMng.getInnerCode();
									String errorMsg = errorMng.getErrMsg();
									int reportflag = sr.getReportFlag();
									String srvNode = sr.getSrvNode();
									String cacheKey = sr.getOrgSerial() + ":" + sr.getPhoneNumber();
									
									if ((0 == reportflag && 0 != errorCode) || (1 == reportflag)) { //reportflag = 0 : 该条消息只有最后出错时要返回状态报告 ; reportflag = 1 该条消息无论最后是否成功都要返回状态报告
											  
										SvrNodeDef svrNodeDef = AppStaticConfigs.svrNodeDefMap.get(srvNode);
										
										if (svrNodeDef != null) {
											
											String acceptReport = svrNodeDef.getAcceptReport();
											
											if (StringUtils.equals(acceptReport, "0") || acceptReport == null) { // 是否向业务端推送报告, 0 推送 1 不推送 2 强制不推送(由于多次连接不成功导致)
												
												SmsClient svrNodeClient = SvrNodeClientsManager.getInstance().getSvrNodeClient(srvNode);
												
												if (svrNodeClient != null) {
													
													String upflag = svrNodeDef.getUpflag(); // 新旧客户端标示
													
													String sendport = StringUtils.toString(sr.getSendPort());
													
													IBody report = null;
													
													if (StringUtils.equals("1", upflag) || StringUtils.equals("2", upflag)) {
														
														long commandId = 0;
														
														if (StringUtils.equals("1", upflag)) {
															report = Report2.build(sr.getOrgSerialPart1(), sr.getOrgSerialPart2(), sr.getOrgSerialPart3(), 
																	sr.getNotifyType(), sr.getPhoneNumber(), errorCode, errorMsg, 
																	sendport,
																	sr.getOrgSrvNode(), sr.getOrgSerialNo());
															commandId = CommandID.SGIP_REPORT2;
														} else {
															report = Report.build(sr.getOrgSerialPart1(), sr.getOrgSerialPart2(), 
																	sr.getOrgSerialPart3(), sr.getNotifyType(), sr.getPhoneNumber(), errorCode, errorMsg);
															commandId = CommandID.SGIP_REPORT;
														}
														
														Head head = Head.build(report.length(), commandId, AppStaticConfigs.smspfmServerCode, StringUtils.toIntegerP(DateTimeUtils.dateToStr(new Date(), "MMddHHmmss")), SerialUtils.getSerial());
														
														NotifyHash notifyHash = new NotifyHash();
														notifyHash.setiSMCacheKey(cacheKey);
														notifyHash.setiRptSnPart1(head.getSerialNoPart1());
														notifyHash.setiRptSnPart2(head.getSerialNoPart2());
														notifyHash.setiRptSnPart3(head.getSerialNoPart3());
														notifyHash.setiSendTimes(1);
														notifyHash.setiNotifyType(sr.getNotifyType());
														notifyHash.setiLastSendTime(new Date());
														
														nfhs.add(notifyHash);
														
														Map<byte[], byte[]> hashes = new HashMap<>();
														if (sr.getNotifyType() == 1) {
															hashes.put(AppCanstant.SmsEntity.ReportTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
															hashes.put(AppCanstant.SmsEntity.ReportSeq, AppStaticConfigs.stringRedisSerializer.serialize(head.getSerialNo()));
														} else {
															hashes.put(AppCanstant.SmsEntity.Notify_send_time, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
															hashes.put(AppCanstant.SmsEntity.SendFeedSeq, AppStaticConfigs.stringRedisSerializer.serialize(head.getSerialNo()));
														}
														connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey), hashes);
														
														
														Message message = new Message(head, report);
														
														logger.reportd(message.toString());
														
														boolean sendReport = svrNodeClient.sendMsg(message);
														
														if (!sendReport) {
															logger.rfailed(sr.toString());
														}
														
													} else {
														
														String desc = "客户端标识upflag为空:";
														
														Map<byte[], byte[]> hashes = new HashMap<>();
														if (sr.getNotifyType() == 1) {
															hashes.put(AppCanstant.SmsEntity.ReportTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
															hashes.put(AppCanstant.SmsEntity.ReportResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
															hashes.put(AppCanstant.SmsEntity.ReportResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
														} else {
															hashes.put(AppCanstant.SmsEntity.Notify_send_time, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
															hashes.put(AppCanstant.SmsEntity.SendFeedbackResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
															hashes.put(AppCanstant.SmsEntity.SendFeedbackResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
														}
														connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey), hashes);
													
														connection.lPush(AppStaticConfigs.iSMNotifyEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey + ":" + sr.getNotifyType()));
														
														logger.rfailed(desc + sr.toString());
													}
													
												} else {
													String desc = "根据SrvNode编号" + sr.getSrvNode() + "创建客户端失败:";
															
													Map<byte[], byte[]> hashes = new HashMap<>();
													if (sr.getNotifyType() == 1) {
														hashes.put(AppCanstant.SmsEntity.ReportTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
														hashes.put(AppCanstant.SmsEntity.ReportResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
														hashes.put(AppCanstant.SmsEntity.ReportResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
													} else {
														hashes.put(AppCanstant.SmsEntity.Notify_send_time, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
														hashes.put(AppCanstant.SmsEntity.SendFeedbackResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
														hashes.put(AppCanstant.SmsEntity.SendFeedbackResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
													}
													connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey), hashes);
												
													connection.lPush(AppStaticConfigs.iSMNotifyEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey + ":" + sr.getNotifyType()));
													
													logger.rfailed(desc + sr.toString());
												}
											} else {
												
												if (StringUtils.equals(acceptReport, "2")) {
													logger.rfailed("业务端:" + svrNodeDef.getSvrNode() + "当前ACCEPT_REPORT状态为2,业务端连接不可用,直接写状态报告通知完成队列, 报告内容:" + sr.toString());
												}
												
												// 直接写通知完成队列
												AppStaticConfigs.listOpts3.in(AppStaticConfigs.iSMNotifyEndList, cacheKey + ":" + sr.getNotifyType());
											}
										} else {
											String desc = "没有匹配到SrvNode编号" + sr.getSrvNode() + ":";
													
											Map<byte[], byte[]> hashes = new HashMap<>();
											if (sr.getNotifyType() == 1) {
												hashes.put(AppCanstant.SmsEntity.ReportTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
												hashes.put(AppCanstant.SmsEntity.ReportResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
												hashes.put(AppCanstant.SmsEntity.ReportResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
											} else {
												hashes.put(AppCanstant.SmsEntity.Notify_send_time, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
												hashes.put(AppCanstant.SmsEntity.SendFeedbackResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
												hashes.put(AppCanstant.SmsEntity.SendFeedbackResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
											}
											connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey), hashes);
										
											connection.lPush(AppStaticConfigs.iSMNotifyEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey + ":" + sr.getNotifyType()));
											
											logger.rfailed(desc + sr.toString());
										}
		
									} else { // 2 该条消息不需要返回状态报告
										
										// 直接写通知完成队列
										AppStaticConfigs.listOpts3.in(AppStaticConfigs.iSMNotifyEndList, cacheKey + ":" + sr.getNotifyType());
									}
									
								} else {
									logger.rfailed("读取的状态报告信息不全:" + sr.toString());
								}
							}
							
							if (AppStaticConfigs.speedStatisticsOn) {
								speeds(entities.size());
							}
						}
					});
					
					List<String> serialnoList = new ArrayList<>();
					
					// 保存重发报告需要到的信息
					AppStaticConfigs.pipelineOpts2.save(new ABExcuteCommands() {

						@Override
						public void excute(RedisConnection connection) {
							
							for (NotifyHash nfh:nfhs) {
								
								String hkey = AppStaticConfigs.pSMNotifySerialNo + ":" + nfh.getSerialNo();
								
								byte[] hkeyb = AppStaticConfigs.stringRedisSerializer.serialize(hkey);
								
								connection.hMSet(hkeyb, nfh.getMapB());
								connection.expire(hkeyb, 86400);
								
								serialnoList.add(nfh.getSerialNo() + ":" + (System.currentTimeMillis() + AppStaticConfigs.rptResendInterval));
							}
						}
					});
					
					// 保存状态报告序列号的同时判断是否收到响应消息 
					List<List<String>> nfhstrlist = AppStaticConfigs.scriptOpts2.getHashList(script2, keys2, serialnoList.toArray());
					
					List<NotifyHash> nfhentities = new ArrayList<NotifyHash>();
					for (int i = 0 ; i < nfhstrlist.size() ; i ++) {
						List<String> tmplist = nfhstrlist.get(i);
						NotifyHash entity = NotifyHash.parseStr(tmplist);
						nfhentities.add(entity);
					}
					
					if (!nfhentities.isEmpty()) {
						
						AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {

							@Override
							public void excute(RedisConnection connection) {
								
								List<byte[]> notifyendList = new ArrayList<>();
								
								for (NotifyHash nfh:nfhentities) {
									
									int iNotifyType = nfh.getiNotifyType();
									
									Map<byte[], byte[]> map = new HashMap<>();
									
									if (iNotifyType == 1) {
										map.put(AppCanstant.SmsEntity.ReportResult, AppStaticConfigs.stringRedisSerializer.serialize("0"));
										map.put(AppCanstant.SmsEntity.ReportResultDesc, AppStaticConfigs.stringRedisSerializer.serialize("success"));
									} else {
										map.put(AppCanstant.SmsEntity.SendFeedbackResult, AppStaticConfigs.stringRedisSerializer.serialize("0"));
										map.put(AppCanstant.SmsEntity.SendFeedbackResultDesc, AppStaticConfigs.stringRedisSerializer.serialize("success"));
									}
									
									connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + nfh.getiSMCacheKey()), map);
									
									notifyendList.add(AppStaticConfigs.stringRedisSerializer.serialize(nfh.getiSMCacheKey() + ":" + iNotifyType));
								}
								
								if (!notifyendList.isEmpty()) {
									byte[][] bs = notifyendList.toArray(new byte[1][]);
									connection.lPush(AppStaticConfigs.iSMNotifyEndListB, bs);
								}
							}
						});
					}
				}
			});
			
			try {
				if (toSleep) {
					Thread.sleep(sleepTime);
				}
			} catch (InterruptedException e) {
				break;
			}
		}
		
		if (executorService != null) {
			executorService.shutdown();
		}
		
		logger.info("状态报告推送任务退出!");
	}
	
	public void speedWriter() {
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while (!executorService.isShutdown()) {
					try {
						//AppStaticConfigs.hashOpts.put(AppStaticConfigs.RunningParamsHash, speeds.take());
						Map<byte[],byte[]> map = speeds.take();
						AppStaticConfigs.pipelineOpts.save(new ABExcuteCommands() {

							@Override
							public void excute(RedisConnection connection) {
								connection.hMSet(AppStaticConfigs.RunningParamsHashB, map);
								// 每秒的速率队列
								byte[] bytes = map.get(CURR_REPORT_SPEEDS_B);
								if (bytes != null) {
									connection.lPush(currReportSpeedS_List, bytes); 
									connection.lTrim(currReportSpeedS_List, 0, AppStaticConfigs.speedSLength - 1);
								}
								// 每分的速率队列
								byte[] bytem = map.get(CURR_REPORT_SPEEDM_B);
								if (bytem != null) {
									connection.lPush(currReportSpeedM_List, bytem); 
									connection.lTrim(currReportSpeedM_List, 0, AppStaticConfigs.speedMLength - 1);
								}
							}
						});
					} catch (InterruptedException e) {
						logger.warn("读speeds错误:"+e.toString());
					}
				}
			}
		}).start();
	}
	
	public synchronized void speeds(int count) {
		
		lastCount += count;
		mLastCount += count;
		
		long currentTimeMillis = System.currentTimeMillis();
		
		if ((currentTimeMillis - lastTime) >= 1000) {
			logger.debug("状态报告回写速度(条/s):" + lastCount);
			
			try {
				String dateTime = DateTimeUtils.dateToStr2(null);
				Map<byte[],byte[]> map = new HashMap<>();
				map.put(CURR_REPORT_SPEEDS_B, AppStaticConfigs.stringRedisSerializer.serialize(lastCount + ":" + dateTime));
				//map.put(REPORT_SPEED_TIMES_B, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(null)));
				speeds.add(map);
			} catch (IllegalStateException  e) {
				logger.error("状态报告回写速率统计队列已满");
			} catch (Exception  e) {
				logger.error(e.getMessage());
			}
			
			lastCount = 0;
			lastTime = System.currentTimeMillis();
		}
		
		if ((currentTimeMillis - mLastTime) >= 60000) {
			logger.debug("状态报告回写速度(条/m):" + lastCount);
			
			try {
				String dateTime = DateTimeUtils.dateToStr2(null);
				Map<byte[],byte[]> map = new HashMap<>();
				map.put(CURR_REPORT_SPEEDM_B, AppStaticConfigs.stringRedisSerializer.serialize(mLastCount + ":" + dateTime));
				//map.put(REPORT_SPEED_TIMEM_B, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(null)));
				speeds.add(map);
			} catch (IllegalStateException  e) {
				logger.error("状态报告回写速率统计队列已满");
			} catch (Exception  e) {
				logger.error(e.getMessage());
			}
			
			mLastCount = 0;
			mLastTime = System.currentTimeMillis();
		}
	}
}
